
package com.naza.rpc.netty;

import com.naza.rpc.async.RpcCallback;
import com.naza.rpc.circuitBreaker.CircuitBreaker;
import com.naza.rpc.circuitBreaker.CircuitBreakerLoader;
import com.naza.rpc.core.MessageCallBack;
import com.naza.rpc.model.RpcRequest;
import com.naza.rpc.model.RpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.net.SocketAddress;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Rpc客户端处理模块  一个server服务器对应一个
 * 
 * @author yl
 */
public class RpcSendHandler extends ChannelInboundHandlerAdapter {

    // requestId,callBack    callback map
    private ConcurrentHashMap<String, MessageCallBack> mapCallBack = new ConcurrentHashMap<String, MessageCallBack>();

    private volatile Channel channel;

    private SocketAddress remoteAddr;

    private CircuitBreakerLoader circuitBreakerLoader = CircuitBreakerLoader.getInstance();

    public Channel getChannel() {
        return channel;
    }

    public SocketAddress getRemoteAddr() {
        return remoteAddr;
    }
    /**
     * 客户端发送器初始化
     * @see io.netty.channel.ChannelInboundHandlerAdapter#channelActive(io.netty.channel.ChannelHandlerContext)
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.remoteAddr = this.channel.remoteAddress();
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    /**
     * 收到服务端返回的响应
     * @see io.netty.channel.ChannelInboundHandlerAdapter#channelRead(io.netty.channel.ChannelHandlerContext, java.lang.Object)
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RpcResponse response = (RpcResponse) msg;
        String messageId = response.getMessageId();
        MessageCallBack callBack = mapCallBack.get(messageId);
        if (callBack != null) {
            mapCallBack.remove(messageId);
            callBack.over(response);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }
    
    /**
     * 向服务端发送请求
     * @param request
     * @return
     */
    public MessageCallBack sendRequest(RpcRequest request) {
        CircuitBreaker circuitBreaker = circuitBreakerLoader.getRetryCircuitBreaker(request.getMethodName());
        MessageCallBack callBack = new MessageCallBack(request,circuitBreaker);
        mapCallBack.put(request.getMessageId(), callBack);
        if(circuitBreaker.isCallPermitted()){
            //向服务端发送请求
            channel.writeAndFlush(request);
        }
        return callBack;
    }

    public MessageCallBack sendRequest(RpcRequest request, RpcCallback rpcCallback) {
        CircuitBreaker circuitBreaker = circuitBreakerLoader.getRetryCircuitBreaker(request.getMethodName());
        MessageCallBack callBack = new MessageCallBack(request,circuitBreaker);
        callBack.setRpcCallback(rpcCallback);
        mapCallBack.put(request.getMessageId(), callBack);
        if(circuitBreaker.isCallPermitted()){
            //向服务端发送请求
            channel.writeAndFlush(request);
        }
        return callBack;
    }
}
